package com.sui.bigdata.flink.sql.side;

import com.sui.bigdata.flink.sql.core.Main;
import org.apache.flink.calcite.shaded.com.google.common.base.Charsets;
import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
import org.junit.Test;

import java.net.URLEncoder;
import java.util.List;

/**
 * @author YongChen
 * @date 2020/5/18 17:05
 * @description
 * @email yong_chen@sui.com
 */
public class ssj_transaction {
    @Test
    public void test1() throws Exception {
        String sql = " " +
                "create scalar function NotBlank with com.sui.bigdata.flink.sql.function.scalar.NotBlank;\n" +
                "\n" +
                "CREATE TABLE dwb_ssj_channel_active(\n" +
                "  udid      varchar,\n" +
                "  channel   varchar,\n" +
                "  cid       varchar\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'dwb_ssj_channel_active_2020/06/15',\n" +
                "    topic ='rthbase.dwb_ssj_channel_active',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " \n" +
                "  CREATE TABLE adl_bd_ls_ssj_transaction_side_ADS798 (\n" +
                "    info:qdlx_6 varchar as qdlx_6 ,\n" +
                "    PRIMARY KEY (rowkey), \n" +
                "    PERIOD FOR SYSTEM_TIME \n" +
                ") WITH (\n" +
                "   type = 'hbase',\n" +
                "   zookeeperQuorum = '10.201.7.114:2181,10.201.7.115:2181,10.201.7.113:2181',\n" +
                "   zookeeperParent = '/hbase',\n" +
                "   tableName = 'adl_bd_ls_ssj_transaction_side_ADS798');\n" +
                " \n" +
                "\n" +
                "CREATE TABLE ssj_transaction(\n" +
                "  suid                  varchar,\n" +
                "  create_time           varchar,\n" +
                "  parent_category_name  varchar,\n" +
                "  fmemo                 varchar\n" +
                " )WITH(\n" +
                "    type ='kafka',\n" +
                "    bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "    offsetReset ='latest',\n" +
                "    groupId = 'ssj_transaction_2020/06/15',\n" +
                "    topic ='etl_dws.dws_bill_ssj_transaction',\n" +
                "    parallelism ='1'\n" +
                " );\n" +
                " \n" +
                " \n" +
                "CREATE TABLE tag1(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     udid varchar,\n" +
                "     devssj_acitve_channel_type varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  );  \n" +
                "  \n" +
                "  CREATE TABLE tag2(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     suid varchar,\n" +
                "     userssj_lastest_medical_mark_time varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  ); \n" +
                "  \n" +
                "  CREATE TABLE tag3(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     suid varchar,\n" +
                "     userssj_lastest_travel_mark_time varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  ); \n" +
                "  \n" +
                "  CREATE TABLE tag4(\n" +
                "     hbase_table varchar,  \n" +
                "     udidmd5 varchar,\n" +
                "     suid varchar,\n" +
                "     userssj_lastest_baby_mark_time varchar\n" +
                "  )WITH(\n" +
                "     type ='kafka',\n" +
                "     bootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "     topic ='rttag.realtime.tag_total',\n" +
                "     parallelism ='1'\n" +
                "  ); \n" +
                " \n" +
                " insert into tag1\n" +
                " select \n" +
                "     'ssj_dev_tagging' as hbase_table,\n" +
                "     md5(t1.udid) as udidmd5, \n" +
                "\t t1.udid,\n" +
                "\t case when t2.qdlx_6 ='ASO渠道'    then 'ASO渠道'\n" +
                "\t      when t2.qdlx_6 ='外部合作'   then '外部合作渠道'\n" +
                "\t\t  when t2.qdlx_6 ='运营系统部' then 'QQ运动'\n" +
                "\t\t  else '其他正常渠道' end as devssj_acitve_channel_type\n" +
                " from (select udid ,md5(concat(channel,cid)) as rowkey \n" +
                "       from  dwb_ssj_channel_active ) t1\n" +
                " join adl_bd_ls_ssj_transaction_side_ADS798 t2\n" +
                " on t1.rowkey = t2.rowkey\n" +
                " where NotBlank(t1.udid);\n" +
                "\n" +
                " insert into tag2\n" +
                " select \n" +
                "     'ssj_user_tagging' as hbase_table,\n" +
                "     md5(suid) as rowkey,\n" +
                "     suid  as suid,\n" +
                "     max(create_time) as userssj_lastest_medical_mark_time\n" +
                " from ssj_transaction\n" +
                " where \t  parent_category_name in ('医疗保健')\n" +
                "       or fmemo like '%挂号%'  \n" +
                "       or fmemo like '%药品%'   \n" +
                "       or fmemo like '%医院%'  \n" +
                "       or fmemo like '%打针%'\n" +
                " group by suid;\n" +
                " \t  \n" +
                " insert into tag3\n" +
                " select \n" +
                "     'ssj_user_tagging' as hbase_table,\n" +
                "     md5(suid) as rowkey,\n" +
                "     suid  as suid,\n" +
                "     max(create_time) as userssj_lastest_travel_mark_time\n" +
                " from ssj_transaction\n" +
                " where    parent_category_name in ('行车交通')\n" +
                "       or fmemo like '%机票%'  \n" +
                " \t   or fmemo like '%高铁%'  \n" +
                " \t   or fmemo like '%车票%'  \n" +
                " \t   or fmemo like '%旅游%'  \n" +
                " \t   or fmemo like '%出行%'  \n" +
                " \t   or fmemo like '%酒店%'  \n" +
                " \t   or fmemo like '%住宿%'  \n" +
                " group by suid;\t\n" +
                " \n" +
                " \n" +
                " insert into tag4\n" +
                " select \n" +
                "     'ssj_user_tagging' as hbase_table,\n" +
                "     md5(suid) as rowkey,\n" +
                "     suid  as suid,\n" +
                "     max(create_time) as userssj_lastest_baby_mark_time\n" +
                " from ssj_transaction\n" +
                " where    fmemo like '%宝宝%'  \n" +
                " \t   or fmemo like '%婴儿%'  \n" +
                " \t   or fmemo like '%纸尿片%'\n" +
                " \t   or fmemo like '%早教%'\n" +
                " \t   or fmemo like '%奶嘴%' \n" +
                " group by suid;  \n" +
                "\t  \n";


        exec(sql);
    }

    @Test
    public void test_ssj_dev_tag_devssj_device_jyb_at_first_time() throws Exception {
        String sql ="create scalar function RiskDecrypt with com.sui.bigdata.flink.sql.function.scalar.RiskDecrypt;\n" +
                "create scalar function FinAesDecode with com.sui.bigdata.flink.sql.function.scalar.FinAesDecode;\n" +
                "create scalar function LoanCclDecrypt with com.sui.bigdata.flink.sql.function.scalar.LoanCclDecrypt;\n" +
                "create scalar function NotBlank with com.sui.bigdata.flink.sql.function.scalar.NotBlank;\n" +
                "create scalar function UnixTimestamp with com.sui.bigdata.flink.sql.function.scalar.UnixTimestamp;\n" +
                "create scalar function Md5 with com.sui.bigdata.flink.sql.function.scalar.Md5;" +
                "CREATE TABLE bdl_ssj_operations(\n" +
                "\t\t\t\t\tudid varchar,\n" +
                "\t\t\t\t\ttimestamp varchar as ts,\n" +
                "\t\t\t\t\taoperation varchar,\n" +
                "\t\t\t\t\tetype varchar\n" +
                "\t\t\t\t )WITH(\n" +
                "\t\t\t\t\ttype ='kafka',\n" +
                "\t\t\t\t\tbootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "\t\t\t\t\toffsetReset ='latest',\n" +
                "\t\t\t\t\tgroupid ='vis_rt_tag_devssj_20200628',\n" +
                "\t\t\t\t\ttopic ='lc.mymoney.operations',\n" +
                "\t\t\t\t\tparallelism ='1'\n" +
                "\t\t\t\t );\n" +
                "\t\t\t \n" +
                "\t\t\t\tCREATE TABLE dim_devssj_trans_at_last_time_side(\n" +
                "\t\t\t\t   info:devssj_trans_at_last_time varchar as devssj_trans_at_last_time,\n" +
                "\t\t\t\t   PRIMARY KEY (udidmd5), \n" +
                "\t\t\t\t   PERIOD FOR SYSTEM_TIME\n" +
                "\t\t\t\t)WITH(\n" +
                "\t\t\t\t   type ='hbase',\n" +
                "\t\t\t\t   zookeeperQuorum ='10.201.7.113:2181,10.201.7.114:2181,10.201.7.115:2181',\n" +
                "\t\t\t\t   zookeeperParent ='/hbase',\n" +
                "\t\t\t\t   tableName ='dim_devssj_trans_at_last_time',\n" +
                "\t\t\t\t   parallelism ='1'\n" +
                "\t\t\t\t);\n" +
                "\n" +
                "\t\t\t\tCREATE TABLE ssj_dev_tag_devssj_device_jyb_at_first_time(\n" +
                "\t\t\t\t\tudidmd5 varchar,\n" +
                "\t\t\t\t\tudid varchar,\n" +
                "\t\t\t\t\thbase_table varchar,\n" +
                "\t\t\t\t\tdevssj_device_jyb_at_first_time varchar\n" +
                "\t\t\t\t )WITH(\n" +
                "\t\t\t\t\ttype ='kafka',\n" +
                "\t\t\t\t\tbootstrapServers ='10.201.7.187:9093,10.201.7.188:9093,10.201.7.189:9093,10.201.7.190:9093',\n" +
                "\t\t\t\t\tkafka.acks = 'all',\n" +
                "\t\t\t\t\tkafka.retries = '2',\n" +
                "\t\t\t\t\ttopic ='rttag.realtime.tag_total',\n" +
                "\t\t\t\t\tparallelism ='5'\n" +
                "\t\t\t\t );\n" +
                "\t\t\t\t\n" +
                "\t\t\t\tinsert into ssj_dev_tag_devssj_device_jyb_at_first_time\n" +
                "\t\t\t\tselect t1.udidmd5 as udidmd5, t1.udid as udid,'ssj_dev_tagging' as hbase_table,devssj_device_jyb_at_first_time from\n" +
                "\t\t\t\t(\n" +
                "\t\t\t\t\tselect t1.udidmd5 as udidmd5,t1.udid as udid,devssj_device_jyb_at_first_time from\n" +
                "\t\t\t\t\t(\n" +
                "\t\t\t\t\t\tselect udid, Md5(udid) as udidmd5,UnixTimestamp(cast(substring(ts,1,10) as bigint)) AS devssj_device_jyb_at_first_time\n" +
                "\t\t\t\t\t\tfrom bdl_ssj_operations \n" +
                "\t\t\t\t\t\twhere NotBlank(udid) AND NotBlank(ts) and (aoperation LIKE '首页_记一笔_%' OR aoperation LIKE 'Home_AddExpense_%') AND etype= 'click'\n" +
                "\t\t\t\t\t) t1\n" +
                "\t\t\t\t\tleft join dim_devssj_trans_at_last_time_side t2\n" +
                "\t\t\t\t\ton t1.udidmd5 = t2.udidmd5\n" +
                "\t\t\t\t) t \n" +
                "\t\t\t\t;";

       exec(sql);
    }






    public void exec(String sql) throws Exception {
        List<String> paramList = Lists.newArrayList();
        paramList.add("-sql");
        String exeSql = URLEncoder.encode(sql, Charsets.UTF_8.name());
        paramList.add(exeSql);
        paramList.add("-name");
        paramList.add("xc");
        paramList.add("-mode");
        paramList.add("local");
        paramList.add("-port");
        paramList.add("8081");
        paramList.add("-confProp");
        String conf = "{\"time.characteristic\":\"EventTime\",\"parallelism\":2}";
        String confEncode = URLEncoder.encode(conf, Charsets.UTF_8.name());
        paramList.add(confEncode);

        String[] params = new String[paramList.size()];
        paramList.toArray(params);
        Main.main(params);
    }
}
